package rds

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/cloudwego/hertz/pkg/common/hlog"
	"time"
)

// PubSubUtils Redis发布订阅工具类
type PubSubUtils struct {
	client *Client
}

// NewPubSubUtils 创建新的发布订阅工具实例
func NewPubSubUtils(client *Client) *PubSubUtils {
	return &PubSubUtils{
		client: client,
	}
}

// PublishJSON 发布JSON消息
func (pu *PubSubUtils) PublishJSON(ctx context.Context, channel string, data interface{}) error {
	jsonData, err := json.Marshal(data)
	if err != nil {
		return fmt.Errorf("marshal json error: %v", err)
	}
	
	return pu.client.PubSub.Publish(ctx, channel, string(jsonData))
}

// PublishJSONWithTTL 发布JSON消息并设置过期时间
func (pu *PubSubUtils) PublishJSONWithTTL(ctx context.Context, channel string, data interface{}, ttl time.Duration) error {
	jsonData, err := json.Marshal(data)
	if err != nil {
		return fmt.Errorf("marshal json error: %v", err)
	}
	
	return pu.client.PubSub.PublishWithTTL(ctx, channel, string(jsonData), ttl)
}

// SubscribeWithJSONHandler 订阅并处理JSON消息
func (pu *PubSubUtils) SubscribeWithJSONHandler(ctx context.Context, channel string, handler func(ctx context.Context, data interface{}) error) error {
	jsonHandler := func(ctx context.Context, msg *Message) error {
		var data interface{}
		if err := json.Unmarshal([]byte(msg.Payload), &data); err != nil {
			return fmt.Errorf("unmarshal json error: %v", err)
		}
		return handler(ctx, data)
	}
	
	pu.client.PubSub.AddHandler(channel, jsonHandler)
	return pu.client.PubSub.Subscribe(ctx, channel)
}

// SubscribeWithStructHandler 订阅并处理结构体消息
func (pu *PubSubUtils) SubscribeWithStructHandler(ctx context.Context, channel string, dataType interface{}, handler func(ctx context.Context, data interface{}) error) error {
	structHandler := func(ctx context.Context, msg *Message) error {
		if err := json.Unmarshal([]byte(msg.Payload), dataType); err != nil {
			return fmt.Errorf("unmarshal struct error: %v", err)
		}
		return handler(ctx, dataType)
	}
	
	pu.client.PubSub.AddHandler(channel, structHandler)
	return pu.client.PubSub.Subscribe(ctx, channel)
}

// PSubscribeWithJSONHandler 模式订阅并处理JSON消息
func (pu *PubSubUtils) PSubscribeWithJSONHandler(ctx context.Context, pattern string, handler func(ctx context.Context, channel string, data interface{}) error) error {
	jsonHandler := func(ctx context.Context, msg *Message) error {
		var data interface{}
		if err := json.Unmarshal([]byte(msg.Payload), &data); err != nil {
			return fmt.Errorf("unmarshal json error: %v", err)
		}
		return handler(ctx, msg.Channel, data)
	}
	
	pu.client.PubSub.AddPatternHandler(pattern, jsonHandler)
	return pu.client.PubSub.PSubscribe(ctx, pattern)
}

// PSubscribeWithStructHandler 模式订阅并处理结构体消息
func (pu *PubSubUtils) PSubscribeWithStructHandler(ctx context.Context, pattern string, dataType interface{}, handler func(ctx context.Context, channel string, data interface{}) error) error {
	structHandler := func(ctx context.Context, msg *Message) error {
		if err := json.Unmarshal([]byte(msg.Payload), dataType); err != nil {
			return fmt.Errorf("unmarshal struct error: %v", err)
		}
		return handler(ctx, msg.Channel, dataType)
	}
	
	pu.client.PubSub.AddPatternHandler(pattern, structHandler)
	return pu.client.PubSub.PSubscribe(ctx, pattern)
}

// Broadcast 广播消息到多个频道
func (pu *PubSubUtils) Broadcast(ctx context.Context, channels []string, message interface{}) error {
	for _, channel := range channels {
		if err := pu.client.PubSub.Publish(ctx, channel, message); err != nil {
			hlog.Errorf("broadcast to channel %s error: %v", channel, err)
			return err
		}
	}
	return nil
}

// BroadcastJSON 广播JSON消息到多个频道
func (pu *PubSubUtils) BroadcastJSON(ctx context.Context, channels []string, data interface{}) error {
	jsonData, err := json.Marshal(data)
	if err != nil {
		return fmt.Errorf("marshal json error: %v", err)
	}
	
	for _, channel := range channels {
		if err := pu.client.PubSub.Publish(ctx, channel, string(jsonData)); err != nil {
			hlog.Errorf("broadcast json to channel %s error: %v", channel, err)
			return err
		}
	}
	return nil
}

// FanOut 扇出模式：一个消息发布到多个频道，每个频道有不同的处理器
func (pu *PubSubUtils) FanOut(ctx context.Context, sourceChannel string, targetChannels []string) error {
	handler := func(ctx context.Context, msg *Message) error {
		return pu.Broadcast(ctx, targetChannels, msg.Payload)
	}
	
	pu.client.PubSub.AddHandler(sourceChannel, handler)
	return pu.client.PubSub.Subscribe(ctx, sourceChannel)
}

// FanOutJSON 扇出模式：一个JSON消息发布到多个频道，每个频道有不同的处理器
func (pu *PubSubUtils) FanOutJSON(ctx context.Context, sourceChannel string, targetChannels []string) error {
	handler := func(ctx context.Context, msg *Message) error {
		var data interface{}
		if err := json.Unmarshal([]byte(msg.Payload), &data); err != nil {
			return fmt.Errorf("unmarshal json error: %v", err)
		}
		return pu.BroadcastJSON(ctx, targetChannels, data)
	}
	
	pu.client.PubSub.AddHandler(sourceChannel, handler)
	return pu.client.PubSub.Subscribe(ctx, sourceChannel)
}

// RetryPublish 重试发布消息
func (pu *PubSubUtils) RetryPublish(ctx context.Context, channel string, message interface{}, maxRetries int, retryDelay time.Duration) error {
	var lastErr error
	for i := 0; i < maxRetries; i++ {
		if err := pu.client.PubSub.Publish(ctx, channel, message); err != nil {
			lastErr = err
			hlog.Warnf("publish retry %d/%d failed: %v", i+1, maxRetries, err)
			if i < maxRetries-1 {
				time.Sleep(retryDelay)
			}
			continue
		}
		return nil
	}
	return fmt.Errorf("publish failed after %d retries, last error: %v", maxRetries, lastErr)
}

// RetryPublishJSON 重试发布JSON消息
func (pu *PubSubUtils) RetryPublishJSON(ctx context.Context, channel string, data interface{}, maxRetries int, retryDelay time.Duration) error {
	var lastErr error
	for i := 0; i < maxRetries; i++ {
		if err := pu.PublishJSON(ctx, channel, data); err != nil {
			lastErr = err
			hlog.Warnf("publish json retry %d/%d failed: %v", i+1, maxRetries, err)
			if i < maxRetries-1 {
				time.Sleep(retryDelay)
			}
			continue
		}
		return nil
	}
	return fmt.Errorf("publish json failed after %d retries, last error: %v", maxRetries, lastErr)
} 